package ws.very.util.akka

import akka.remote.testkit.MultiNodeConfig
import akka.remote.testkit.MultiNodeSpec
import akka.testkit.ImplicitSender
import ws.very.util.akka.common.STMultiNodeSpec
import com.typesafe.config.ConfigFactory
import akka.cluster.Cluster
import akka.cluster.ClusterEvent.MemberUp
import akka.cluster.ClusterEvent.CurrentClusterState
import ws.very.util.akka.cluster._
import ws.very.util.akka.msg.MsgCenter
import ws.very.util.el.Els
import scala.concurrent.duration._
import akka.testkit.TestProbe

object Config extends MultiNodeConfig {
  val node1 = role("node1")
  val node2 = role("node2")
  val node3 = role("node3")
  val node4 = role("node4")
  val node5 = role("node5")
  commonConfig(ConfigFactory.parseString {
    """
akka {
  loglevel = "INFO"
  actor {
    provider = "akka.cluster.ClusterActorRefProvider"
    debug.fsm = on
  }
  remote {
    log-remote-lifecycle-events = off
    netty.tcp {
      hostname = "127.0.0.1"
      port = 0
    }
  }

  cluster {
    seed-nodes = []

    auto-down-unreachable-after = 10s
  }
}
    """
  })
}
class ClusterMsgCenter extends MultiNodeSpec(Config)
  with STMultiNodeSpec with ImplicitSender {
  import akka.pattern.ask
  import scala.concurrent.ExecutionContext.Implicits.global
  def initialParticipants = roles.size
  "ClusterMsgCenter" must {
    val ac = ClusterMsgCenter(system)

    val cluster = Cluster(system)
    "all nodes join cluster and reg" in {

      cluster.join(node(Config.node1).address)
      cluster.registerOnMemberUp {

        ac ! ClusterReplication.Send2All(MsgCenter.RegForward(myself.name, Els.Eq("test" + myself.name), self, 123))
      }

      expectMsg(35.seconds, true)
      testConductor.enter("all-up")
    }

    "all regs is same" in {

      //    ( system.actorSelection(MsgCenter.name) ? MsgCenter.CurRegs)(10 seconds).mapTo[Iterable[MsgCenter.RegForward]].map {it=> it.map{_.id} }
      cluster.registerOnMemberUp {
        this.synchronized { this.wait(5000) } //等群集信息同步
        system.actorSelection("/user/" + MsgCenter.name) ! MsgCenter.CurRegs
      }
      expectMsgPF(50 seconds) {
        case t: Iterable[MsgCenter.RegForward] if t.map { _.id }.toSet == ((1 to initialParticipants).map { "node"+ }).toSet => println("??????????????" + t.map { _.id })
      }
      testConductor.enter("check same end")
      //      this.synchronized { this.wait(30000) }//给别人发同步消息
    }

    "req is right" in {
      (1 to initialParticipants).foreach { i =>
        system.actorSelection("/user/" + MsgCenter.name) ! ("testnode" + i)
      }
      (1 to initialParticipants).foreach { i =>
        expectMsg(35.seconds, "test" + myself.name)
      }

      testConductor.enter("req end")
    }

  }
}

class ClusterMsgCenterMultiJvmNode1 extends ClusterMsgCenter
class ClusterMsgCenterMultiJvmNode2 extends ClusterMsgCenter
class ClusterMsgCenterMultiJvmNode3 extends ClusterMsgCenter
class ClusterMsgCenterMultiJvmNode4 extends ClusterMsgCenter
class ClusterMsgCenterMultiJvmNode5 extends ClusterMsgCenter